/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce;

import java.io.IOException;
import java.util.Map;
import java.util.HashMap;
import org.msgpack.MessagePack;
import org.msgpack.packer.BufferPacker;
import org.msgpack.unpacker.BufferUnpacker;

import org.sss.mapreduce.datatype.Packable;

import org.sss.mapreduce.datatype.PackableNull;
import org.sss.mapreduce.datatype.PackableBytes;
import org.sss.mapreduce.datatype.PackableDouble;
import org.sss.mapreduce.datatype.PackableFloat;
import org.sss.mapreduce.datatype.PackableInt;
import org.sss.mapreduce.datatype.PackableLong;
import org.sss.mapreduce.datatype.PackableString;
import org.sss.mapreduce.datatype.TimedPackableLong;
import org.sss.mapreduce.datatype.TimedPackableString;

public class BasicTypeEncoding implements Encoding {
  private static MessagePack msgpack = new MessagePack();

  private interface Factory {
    Encoder createEncoder() throws SssException;
    Decoder createDecoder() throws SssException;
  };
  private Map<Class<? extends Packable>, Factory> factoryMap = new HashMap<Class<? extends Packable>, Factory>();

  private BasicTypeEncoding() {
    factoryMap.put(PackableNull.class,        new PackableNullFactory());
    factoryMap.put(PackableBytes.class,       new PackableBytesFactory());
    factoryMap.put(PackableDouble.class,      new PackableDoubleFactory());
    factoryMap.put(PackableFloat.class,       new PackableFloatFactory());
    factoryMap.put(PackableInt.class,         new PackableIntFactory());
    factoryMap.put(PackableLong.class,        new PackableLongFactory());
    factoryMap.put(PackableString.class,      new PackableStringFactory());
    factoryMap.put(TimedPackableLong.class,   new TimedPackableLongFactory());
    factoryMap.put(TimedPackableString.class, new TimedPackableStringFactory());
  }

  @Override
  public boolean canEncode(Class<? extends Packable> klass) throws SssException {
    return factoryMap.containsKey(klass);
  }

  @Override
  public Encoder createEncoder(Class<? extends Packable> klass) throws SssException {
    assert canEncode(klass): "invalid type";
    return factoryMap.get(klass).createEncoder();
  }

  @Override
  public Decoder createDecoder(Class<? extends Packable> klass) throws SssException {
    assert canEncode(klass): "invalid type";
    return factoryMap.get(klass).createDecoder();
  }

  private static byte[] EMPTY_BYTES = new byte[0];
  private static class PackableNullFactory implements Factory {
    public Encoder createEncoder() throws SssException {
      return new Encoder() {
        public byte[] encode(Packable p) throws IOException {
          @SuppressWarnings("unused")
          PackableNull pn = ((PackableNull)p);
          return EMPTY_BYTES;
        }
      };
    }
    public Decoder createDecoder() throws SssException {
      return new Decoder() {
        public Packable decode(byte[] b) throws IOException {
          if (b.length != 0) {
            throw new IOException("PackableNull Decoder: The byte array length is not 0.");
          }
          return new PackableNull();
        }
      };
    }
  }

  private static class PackableBytesFactory implements Factory {
    public Encoder createEncoder() throws SssException {
      return new Encoder() {
        public byte[] encode(Packable p) throws IOException {
          return ((PackableBytes)p).get();
        }
      };
    }
    public Decoder createDecoder() throws SssException {
      return new Decoder() {
        public Packable decode(byte[] b) throws IOException {
          return new PackableBytes(b);
        }
      };
    }
  }

  private static class PackableDoubleFactory implements Factory {
    public Encoder createEncoder() throws SssException {
      final BufferPacker packer = msgpack.createBufferPacker();
      return new Encoder() {
        public byte[] encode(Packable p) throws IOException {
          try {
            packer.write(((PackableDouble)p).get());
            return packer.toByteArray();
          }
          finally {
            packer.clear();
          }
        }
      };
    }
    public Decoder createDecoder() throws SssException {
      final BufferUnpacker unpacker = msgpack.createBufferUnpacker();
      return new Decoder() {
        public Packable decode(byte[] b) throws IOException {
          unpacker.wrap(b);
          return new PackableDouble(unpacker.readDouble());
        }
      };
    }
  }

  private static class PackableFloatFactory implements Factory {
    public Encoder createEncoder() throws SssException {
      final BufferPacker packer = msgpack.createBufferPacker();
      return new Encoder() {
        public byte[] encode(Packable p) throws IOException {
          try {
            packer.write(((PackableFloat)p).get());
            return packer.toByteArray();
          }
          finally {
            packer.clear();
          }
        }
      };
    }
    public Decoder createDecoder() throws SssException {
      final BufferUnpacker unpacker = msgpack.createBufferUnpacker();
      return new Decoder() {
        public Packable decode(byte[] b) throws IOException {
          unpacker.wrap(b);
          return new PackableFloat(unpacker.readFloat());
        }
      };
    }
  }

  private static class PackableIntFactory implements Factory {
    public Encoder createEncoder() throws SssException {
      final BufferPacker packer = msgpack.createBufferPacker();
      return new Encoder() {
        public byte[] encode(Packable p) throws IOException {
          try {
            packer.write(((PackableInt)p).get());
            return packer.toByteArray();
          }
          finally {
            packer.clear();
          }
        }
      };
    }
    public Decoder createDecoder() throws SssException {
      final BufferUnpacker unpacker = msgpack.createBufferUnpacker();
      return new Decoder() {
        public Packable decode(byte[] b) throws IOException {
          unpacker.wrap(b);
          return new PackableInt(unpacker.readInt());
        }
      };
    }
  }

  private static class PackableLongFactory implements Factory {
    public Encoder createEncoder() throws SssException {
      final BufferPacker packer = msgpack.createBufferPacker();
      return new Encoder() {
        public byte[] encode(Packable p) throws IOException {
          try {
            packer.write(((PackableLong)p).get());
            return packer.toByteArray();
          }
          finally {
            packer.clear();
          }
        }
      };
    }
    public Decoder createDecoder() throws SssException {
      final BufferUnpacker unpacker = msgpack.createBufferUnpacker();
      return new Decoder() {
        public Packable decode(byte[] b) throws IOException {
          unpacker.wrap(b);
          return new PackableLong(unpacker.readLong());
        }
      };
    }
  }

  private static class PackableStringFactory implements Factory {
    public Encoder createEncoder() throws SssException {
      final BufferPacker packer = msgpack.createBufferPacker();
      return new Encoder() {
        public byte[] encode(Packable p) throws IOException {
          try {
            packer.write(((PackableString)p).get());
            return packer.toByteArray();
          }
          finally {
            packer.clear();
          }
        }
      };
    }
    public Decoder createDecoder() throws SssException {
      final BufferUnpacker unpacker = msgpack.createBufferUnpacker();
      return new Decoder() {
        public Packable decode(byte[] b) throws IOException {
          unpacker.wrap(b);
          return new PackableString(unpacker.readString());
        }
      };
    }
  }

  private static class TimedPackableLongFactory implements Factory {
    public Encoder createEncoder() throws SssException {
      final BufferPacker packer = msgpack.createBufferPacker();
      return new Encoder() {
        public byte[] encode(Packable p) throws IOException {
          try {
            TimedPackableLong l = (TimedPackableLong)p;
            packer.write(l.getLong());
            packer.write(l.getTime());
            return packer.toByteArray();
          }
          finally {
            packer.clear();
          }
        }
      };
    }
    public Decoder createDecoder() throws SssException {
      final BufferUnpacker unpacker = msgpack.createBufferUnpacker();
      return new Decoder() {
        public Packable decode(byte[] b) throws IOException {
          unpacker.wrap(b);
          long value = unpacker.readLong();
          long time  = unpacker.readLong();
          return new TimedPackableLong(value, time);
        }
      };
    }
  }

  private static class TimedPackableStringFactory implements Factory {
    public Encoder createEncoder() throws SssException {
      final BufferPacker packer = msgpack.createBufferPacker();
      return new Encoder() {
        public byte[] encode(Packable p) throws IOException {
          try {
            TimedPackableString s = (TimedPackableString)p;
            packer.write(s.getString());
            packer.write(s.getTime());
            return packer.toByteArray();
          }
          finally {
            packer.clear();
          }
        }
      };
    }
    public Decoder createDecoder() throws SssException {
      final BufferUnpacker unpacker = msgpack.createBufferUnpacker();
      return new Decoder() {
        public Packable decode(byte[] b) throws IOException {
          unpacker.wrap(b);
          String value = unpacker.readString();
          long time  = unpacker.readLong();
          return new TimedPackableString(value, time);
        }
      };
    }
  }

  public static BasicTypeEncoding getInstance() {
    return instance;
  }

  private static BasicTypeEncoding instance = new BasicTypeEncoding();
}
